package com.teamsun.mqtt.handle;

import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubAckPayload;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.service.GetUnReadData;

/**
 * 处理订阅的handle
 * 
 * @author acer
 *
 */
@Sharable
public class SubServiceHandle extends ChannelInboundHandlerAdapter {

	Logger logger = Logger.getLogger(getClass());

	ConcurrentHashMap<Channel, String> channel2str;
	/**
	 * 客户端订阅表
	 */
	ConcurrentNavigableMap<String, BlockingQueue<String>> submap;
	ExecutorService dboptgroup;

	@Autowired
	GetUnReadData getUnReadData;

	public SubServiceHandle(ExecutorService executorService) {
		this.dboptgroup = executorService;
	}

	public void channelRead(ChannelHandlerContext ctx, Object msg) {

		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttMessageType messageType = message.fixedHeader().messageType();

			switch (messageType) {
			case SUBSCRIBE:
				sub(ctx, (MqttSubscribeMessage) message);
				break;
			default:
				ctx.fireChannelRead(msg);
				break;
			}

		}

		else
			ctx.close();
	}

	/**
	 * 订阅操作 向服务器订阅所有感兴趣的主题 并且按照客户端标识找到未读消息接受这些消息
	 * 
	 * @param ctx
	 * @param subscribeMessage
	 */
	private void sub(final ChannelHandlerContext ctx, MqttSubscribeMessage subscribeMessage) {

		MqttQoS mqttQoS = subscribeMessage.fixedHeader().qosLevel();

		MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, mqttQoS, false, 0);

		List<MqttTopicSubscription> list = subscribeMessage.payload().topicSubscriptions();

		List<String> topNames = new ArrayList<String>();
		int[] qoslevels = new int[list.size()];
		int i = 0;
		for (MqttTopicSubscription subscription : list) {
			topNames.add(subscription.topicName());
			qoslevels[i] = subscription.qualityOfService().value();
		}

		if (qoslevels.length > 0) {
			MqttSubAckPayload payload = new MqttSubAckPayload(qoslevels);
			MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader
					.from(subscribeMessage.variableHeader().messageId());
			MqttSubAckMessage subAckMessage = new MqttSubAckMessage(fixedHeader, mqttMessageIdVariableHeader, payload);
			ctx.writeAndFlush(subAckMessage);
		}

		if (!topNames.isEmpty()) {
			String iden = channel2str.get(ctx.channel());

			StringBuilder builder = new StringBuilder();
			for (String topName : topNames) {
				BlockingQueue<String> clientIdens = submap.get(topName);
				if (clientIdens == null) {
					clientIdens = new LinkedBlockingQueue<String>();
					submap.put(topName, clientIdens);
				}
				if (!clientIdens.contains(iden)) {
					clientIdens.offer(iden);
				}

				builder.append(topName + ",");

			}
			logger.debug(iden + "订阅了,主题" + builder.toString());
			submitGetUnReadyMsg(iden,ctx.alloc());
		}

	}

	/**
	 * 在订阅的时候 指定的 client 接受消息
	 * 
	 * @param clientid
	 */
	private void submitGetUnReadyMsg(final String clientid,ByteBufAllocator allocator) {

		dboptgroup.submit(() -> {
			try {
				getUnReadData.sendUnReadMsg(clientid,allocator);
			} catch (Exception e) {
				logger.debug("异常",e);
			}
		});
	}

	@Override
	public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
		ctx.close();
	}

	public void setChannel2str(ConcurrentHashMap<Channel, String> channel2str) {
		this.channel2str = channel2str;
	}

	public void setSubmap(ConcurrentNavigableMap<String, BlockingQueue<String>> submap) {
		this.submap = submap;
	}

}
